package com.xbai.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * WordCount案例
 * 使用 netcat 工具向 9999 端口不断的发送数据，通过 SparkStreaming 读取端口数据并统计不同单词出现的次数
 */
object SparkStreaming01_WordCount {

  def main(args: Array[String]): Unit = {
    // spark 配置对象
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    // 实时数据分析环境对象，采集周期，以指定的时间为周期采集实时数据
    val streamingContext: StreamingContext = new StreamingContext(conf, Seconds(5))

    // 从指定端口采集数据
    val socketLineStream: ReceiverInputDStream[String] = streamingContext.socketTextStream("192.168.115.102", 9999)

    val wordDStream: DStream[String] = socketLineStream.flatMap(line => line.split(" "))

    val mapDStream: DStream[(String, Int)] = wordDStream.map((_, 1))

    val wordToSumDStream: DStream[(String, Int)] = mapDStream.reduceByKey(_+_)

    wordToSumDStream.print()

    // 启动采集器，不能停止采集程序
    streamingContext.start()

    // Driver 等待采集器的执行
    streamingContext.awaitTermination()

  }
}
